package org.xwb.springcloud.messagetype.deadletter;

import com.alibaba.fastjson2.util.DateUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.xwb.springcloud.constant.MqDeadLetterConstant;

import java.io.IOException;
import java.util.Date;
import java.util.Map;

/**
 * @author 10564
 */
@Component
public class DeadLetterBusinessAndRetryCountConsumer {
    private static final Logger log = LoggerFactory.getLogger(DeadLetterBusinessAndRetryCountConsumer.class);
    Map<String, Integer> retryCountMap = new java.util.HashMap<>();

    @Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")
    private Integer retryCount;

    /**
     * @param msg     消息内容
     * @param channel 通道对象
     * @param tag     消息的tag
     */
    @RabbitListener(queues = MqDeadLetterConstant.BUSINESS_QUEUE)
    public void receiveAckBusinessQueueMessage(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        try {
            log.info("\n消费者eadLetter接收消息:【{}】\n", msg);
            if ("normal".equals(msg)) {
                log.info("\n已被正常消费 normal 【{}】,【{}】\n", msg, DateUtils.format(new Date()));
                //todo 消息正常处理，从队列中删除掉当前消息
                channel.basicAck(tag, false);
            }  else {
                //todo 此处模拟消息消费异常，计数器累加，超过5次则拒绝加入队列 正常情况下使用redis 实现计数器，此处为了演示直接使用map代替
                Integer timesObj = retryCountMap.get("msg:" + tag);
                int times = timesObj == null ? 0 : timesObj;
                if (times >= retryCount - 1) {
                    channel.basicNack(tag, false, false);
                } else {
                    times = times == 0 ? 1 : times + 1;
                    retryCountMap.put("msg:" + tag, times);
                    log.error("\n已被消费异常，开始重试【{}】,第【{}】次\n", msg, times);
                    throw new Exception(msg);
                }
            }
        } catch (Exception e) {
            //抛出异常，触发重试机制
            throw e;
        }
    }

    /**
     * 死信队列消费者
     *
     * @param msg     消息
     * @param channel 通道对象
     * @param tag     消息的tag
     */
    @RabbitListener(queues = MqDeadLetterConstant.DEAD_LETTER_QUEUE)
    public void receiveAckDlqQueue(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            // 处理死信队列消息
            log.error("\n死信消费者接收消息:【{}】\n", msg);
            //todo true 允许加入队列， false 拒绝加入队列
            channel.basicAck(tag, false);
        } catch (Exception e) {
            log.error("\n死信消费者消息异常:【{}】\n", e.getMessage());
            try {
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}
